package weaver.splittable.task;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import weaver.conn.ConnectionPool;
import weaver.conn.WeaverConnection;
import weaver.conn.WeaverStatement;
import weaver.splittable.STConfig;
import weaver.splittable.STContext;
import weaver.splittable.db.STDBConstants;
import weaver.splittable.db.STDBUtil;
import weaver.splittable.modal.DataChangeInfo;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;

/**
 * 数据同步任务
 * 将现有表数据变动同步到全量表中
 * TimerTask 执行间隔在分表配置文件中设置 默认5s
 * author: YUEYANG 2023-01-16
 */
public class STSyncTask extends TimerTask {

    private static final Logger LOGGER = Logger.getLogger(STConfig.LOGGER_NAME);

    private static boolean isRunning = false;

    @Override
    public void run() {
        if (!isRunning && !STClearTask.isRunning) {
            try {
                isRunning = true;
                sync();
            } catch (Exception e) {
                LOGGER.error("ST SyncTask error:", e);
            } finally {
                isRunning = false;
            }
        }
    }

    private void sync() throws Exception {

        boolean isFinished = false;
        while (!isFinished) {
            Long startTime = System.currentTimeMillis();
            Integer syncCount = 0;
            WeaverConnection connection = null;
            WeaverStatement statement = null;
            try {
                connection = ConnectionPool.getInstance().getConnection();
                connection.setAutoCommit(false);
                statement = (WeaverStatement) connection.createStatement();

                // 获取数据变动记录 按1000条一次获取
                List<DataChangeInfo> dataChangeInfos = getDataChangeInfos(connection, 1000);
                syncCount = dataChangeInfos.size();
                if (syncCount < 1000) {
                    isFinished = true;
                }

                if (dataChangeInfos.size() > 0) {

                    // 剔除清理现有表导致的数据变动
                    dataChangeInfos = eliminateClearDataChange(connection, dataChangeInfos);

                    // 剔除被覆盖的数据变动 目前测试性能尚可，暂不执行此逻辑
                    // dataChangeInfos = eliminateOverDataChange(connection, dataChangeInfos);

                    // 同步变动到全量表
                    executeChange(connection, dataChangeInfos);

                }
                connection.commit();
            } catch (Exception e) {
                connection.rollback();
                LOGGER.error("ST Sync error:", e);
            } finally {
                if (statement != null) {
                    statement.close();
                }
                if (connection != null) {
                    connection.setAutoCommit(true);
                    ConnectionPool.getInstance().returnConnection(connection);
                }
            }
            Long endTime = System.currentTimeMillis();
            LOGGER.info("ST Sync count: " + syncCount + ", time: " + (endTime - startTime) + "ms");
        }

    }

    /**
     * 一次性同步 执行一次同步 同步所有未同步的数据
     * 如果调用时有同步线程在执行，则放弃同步
     * 此方法执行时，同步任务会停止执行
     * @param connection
     */
    public void onceSync(WeaverConnection connection) {
        try {
            List<DataChangeInfo> dataChangeInfos = getDataChangeInfos(connection, null);
            if (dataChangeInfos.size() > 0) {
                if (!isRunning) {
                    try {
                        isRunning = true;
                        int count = 0;
                        List<DataChangeInfo> executeChanges = new ArrayList<>();
                        for (DataChangeInfo dataChangeInfo: dataChangeInfos) {
                            executeChanges.add(dataChangeInfo);
                            count++;
                            if (count == 1000) {
                                executeChanges = eliminateClearDataChange(connection, executeChanges);
                                executeChanges = eliminateOverDataChange(connection, executeChanges);
                                executeChange(connection, executeChanges);
                                count = 0;
                                executeChanges.clear();
                            }
                        }
                        if (executeChanges.size() > 0) {
                            executeChanges = eliminateClearDataChange(connection, executeChanges);
                            executeChanges = eliminateOverDataChange(connection, executeChanges);
                            executeChange(connection, executeChanges);
                        }
                    } catch (Exception e) {
                        LOGGER.error("STSyncTask onceSync error:", e);
                    } finally {
                        isRunning = false;
                    }
                }
            }
        } catch (Exception e) {
            LOGGER.error("STSyncTask onceSync getDataChangeInfos error:", e);
        }
    }

    /**
     * 剔除由于清理活跃表产生的数据变动
     * @param connection 数据库连接
     * @param dataChangeInfos 全部数据变动
     * @return 剔除后的数据变动
     * @throws Exception
     */
    private List<DataChangeInfo> eliminateClearDataChange(WeaverConnection connection, List<DataChangeInfo> dataChangeInfos) throws Exception {
        // 获取Delete变动的记录
        List<DataChangeInfo> deleteChangeInfos = new ArrayList<>();
        for (DataChangeInfo dataChangeInfo: dataChangeInfos) {
            if ("delete".equals(dataChangeInfo.getType())) {
                deleteChangeInfos.add(dataChangeInfo);
            }
        }

        // 如果有delete变动
        if (deleteChangeInfos.size() > 0) {

            Map<String, Map<String, Long>> dataClearInfoCache = new HashMap<>();

            // 去除现有表清理记录
            List<DataChangeInfo> clearDataChanges = new ArrayList<>();
            List<Long> clearDataIds = new ArrayList<>();
            List<Long> clearDataChangeIds = new ArrayList<>();
            for (DataChangeInfo dataChangeInfo: deleteChangeInfos) {
                String tableName = dataChangeInfo.getTableName();
                String dataId = dataChangeInfo.getDataId();
                Long id = dataChangeInfo.getId();

                // 获取当前表的未同步清理数据 先从cache获取，cache没有从数据库获取并放入cache中
                // 如果未同步清理表数据有上百万 这里可能导致内存溢出，考虑不使用缓存直接循环查表（也可能有性能问题）
                // 2023-02-06 修改清理表清理逻辑后，清理完后剩余的清理表数据最多只会有1000条，不会出现上百万的情况
                Map<String, Long> dataClearInfos = null;
                if (dataClearInfoCache.containsKey(dataChangeInfo.getTableName())) {
                    dataClearInfos = dataClearInfoCache.get(tableName);
                } else {
                    dataClearInfos = getDataClearInfos(connection, tableName);
                    dataClearInfoCache.put(tableName, dataClearInfos);
                }

                // 判断当前变动记录是否是清理变动
                if (dataClearInfos != null && dataClearInfos.size() > 0) {
                    if (dataClearInfos.containsKey(dataId)) {
                        clearDataChanges.add(dataChangeInfo);
                        clearDataChangeIds.add(id);
                        clearDataIds.add(dataClearInfos.get(dataId));
                    }
                }
            }
            if (clearDataChanges.size() > 0) {
                WeaverStatement statement = (WeaverStatement) connection.createStatement();
                try {
                    LOGGER.info("剔除清理数据变动：" + clearDataChanges.size());
                    deleteChangeInfos.removeAll(clearDataChanges);
                    dataChangeInfos.removeAll(clearDataChanges);
                    String clearDataInfoIdStr = StringUtils.join(clearDataIds.toArray(), ",");
                    statement.originExecuteUpdate("update "+ STDBConstants.DATA_CLEAR_TABLE_NAME +" set synced = '1', sync_time = "+ System.currentTimeMillis() +" where id in ( " + clearDataInfoIdStr + " ) ");

                    String dataChangeInfoIdStr = StringUtils.join(clearDataChangeIds.toArray(), ",");
                    statement.originExecuteUpdate("update "+ STDBConstants.DATA_CHANGE_TABLE_NAME +" set synced='4', sync_time = "+ System.currentTimeMillis() +" where id in (" + dataChangeInfoIdStr + ")");
                }catch (Exception e) {
                    throw e;
                } finally {
                    if (statement != null) {
                        statement.close();
                    }
                }
            }
        }
        return dataChangeInfos;
    }

    /**
     * 剔除覆盖的操作，减少无意义的数据库交互
     * @param connection 数据库连接
     * @param dataChangeInfos 全部数据变动记录
     * @return 剔除后的数据变动记录
     * @throws Exception
     */
    private List<DataChangeInfo> eliminateOverDataChange(WeaverConnection connection, List<DataChangeInfo> dataChangeInfos) throws Exception {
        // 按表名进行分类
        Map<String, List<DataChangeInfo>> dataChangeInfoMapForTableName = new HashMap<>();
        for (DataChangeInfo dataChangeInfo: dataChangeInfos) {
            String tableName = dataChangeInfo.getTableName();
            if (dataChangeInfoMapForTableName.containsKey(tableName)) {
                dataChangeInfoMapForTableName.get(tableName).add(dataChangeInfo);
            } else {
                List<DataChangeInfo> dataChangeInfoList = new ArrayList<>();
                dataChangeInfoList.add(dataChangeInfo);
                dataChangeInfoMapForTableName.put(tableName, dataChangeInfoList);
            }
        }

        // 去除被覆盖的操作 覆盖的synced值为3
        List<Long> overChangeIds = new ArrayList<>(); // 被覆盖的变动ID
        Set<DataChangeInfo> executeChanges = new HashSet<>(); // 需要执行的变动
        for (Map.Entry<String, List<DataChangeInfo>> entry: dataChangeInfoMapForTableName.entrySet()) {
            // 按dataId 进行分组后按创建时间排序
            // 如果最后一个变动是删除 则判断前面操作是否有插入，有则包括删除全部覆盖，没有则除了删除全部覆盖 未实现
            // 如果最后一个变动是修改 则判断前面操作是否有插入，有则需要执行插入和修改，中间操作全部覆盖，没有则除了最后修改全部覆盖 未实现
            // 目前逻辑是把同一条数据的操作按时间排序，同样操作新的覆盖旧的
            List<DataChangeInfo> dataChangeInfoList = entry.getValue();

            HashMap<String, List<DataChangeInfo>> dataChangeInfoMapForDataId = new HashMap<>();
            for (DataChangeInfo dataChangeInfo: dataChangeInfoList) {
                String dataId = dataChangeInfo.getDataId();
                if (dataChangeInfoMapForDataId.containsKey(dataId)) {
                    dataChangeInfoMapForDataId.get(dataId).add(dataChangeInfo);
                } else {
                    List<DataChangeInfo> dataChangeInfoListForDataId = new ArrayList<>();
                    dataChangeInfoListForDataId.add(dataChangeInfo);
                    dataChangeInfoMapForDataId.put(dataId, dataChangeInfoListForDataId);
                }
            }

            for (Map.Entry<String, List<DataChangeInfo>> entryForDataId: dataChangeInfoMapForDataId.entrySet()) {
                List<DataChangeInfo> datachangeInfoListForDataId = entryForDataId.getValue();
                // 同一条数据超过2次变动才需要判断是否有覆盖变动
                if (datachangeInfoListForDataId.size() < 2) {
                    executeChanges.add(datachangeInfoListForDataId.iterator().next());
                    continue;
                }

                // 需要按创建时间进行排序 下面覆盖相同操作时，保证最新的操作留下来
                Collections.sort(datachangeInfoListForDataId);

                // 按操作分组 覆盖相同的操作
                HashMap<String, DataChangeInfo> dataChangeInfoMapForType = new HashMap<>();
                for (DataChangeInfo dataChangeInfo: datachangeInfoListForDataId) {
                    String type = dataChangeInfo.getType();
                    if (dataChangeInfoMapForType.containsKey(type)) {
                        DataChangeInfo oldDataChangeInfo = dataChangeInfoMapForType.get(type);
                        overChangeIds.add(oldDataChangeInfo.getId()); // 记录被覆盖的操作

                        if (executeChanges.contains(oldDataChangeInfo)) {
                            executeChanges.remove(oldDataChangeInfo);
                        }
                    }
                    dataChangeInfoMapForType.put(type, dataChangeInfo);
                    executeChanges.add(dataChangeInfo);
                }
            }
        }
        if (overChangeIds.size() > 0) {
            LOGGER.info("剔除冗余的数据变动：" + overChangeIds.size());
            WeaverStatement statement = (WeaverStatement) connection.createStatement();
            try {
                String overChangeIdStr = StringUtils.join(overChangeIds.toArray(), ",");
                statement.originExecuteUpdate("update "+ STDBConstants.DATA_CHANGE_TABLE_NAME +" set synced='3', sync_time = "+ System.currentTimeMillis() +" where id in (" + overChangeIdStr + ")");
            } catch (Exception e) {
                throw e;
            } finally {
                if (statement != null) {
                    statement.close();
                }
            }
        }
        return dataChangeInfos;
    }


    /**
     * 执行变动，将活跃表中的数据变动同步到全量表中
     * @param connection 数据库连接
     * @param dataChangeInfos 待执行的变动
     * @throws Exception
     */
    private void executeChange(WeaverConnection connection, List<DataChangeInfo> dataChangeInfos) throws Exception {
        // 按表名进行分类
        Map<String, List<DataChangeInfo>> dataChangeInfoMapForTableName = new HashMap<>();
        List<Long> executeIds = new ArrayList<>();
        for (DataChangeInfo dataChangeInfo: dataChangeInfos) {

            executeIds.add(dataChangeInfo.getId());

            String tableName = dataChangeInfo.getTableName();
            if (dataChangeInfoMapForTableName.containsKey(tableName)) {
                dataChangeInfoMapForTableName.get(tableName).add(dataChangeInfo);
            } else {
                List<DataChangeInfo> dataChangeInfosList = new ArrayList<>();
                dataChangeInfosList.add(dataChangeInfo);
                dataChangeInfoMapForTableName.put(tableName, dataChangeInfosList);
            }
        }

        // 按操作类型分类后执行
        for (Map.Entry<String, List<DataChangeInfo>> entry: dataChangeInfoMapForTableName.entrySet()) {

            List<String> insertedIds = new ArrayList<>();
            List<String> updatedIds = new ArrayList<>();
            List<String> deletedIds = new ArrayList<>();

            String tableName = entry.getKey();
            String targetTableName = tableName + "_all";
            String dataIdColumn = entry.getValue().iterator().next().getDataIdColumn();
            List<DataChangeInfo> dataChangeInfoList = entry.getValue();

            // INSERT UPDATE DELETE 必须全部分开 不能用delete和insert代替update 有风险
            // 生成SQL语句的时候 有几率出现异常 需要发现原因
            for (DataChangeInfo dataChangeInfo: dataChangeInfoList) {
                if (dataChangeInfo.getType().equals("insert")) {
                    insertedIds.add(dataChangeInfo.getDataId());
                } else if (dataChangeInfo.getType().equals("update")) {
                    updatedIds.add(dataChangeInfo.getDataId());
                } else if (dataChangeInfo.getType().equals("delete")) {
                    deletedIds.add(dataChangeInfo.getDataId());
                }
            }

            if (deletedIds.size() > 0) {
                STDBUtil.deleteDataForId(targetTableName, deletedIds, connection, dataIdColumn);
            }
            if (insertedIds.size() > 0) {
                STDBUtil.insertBatch(tableName, targetTableName, insertedIds, connection, dataIdColumn);
            }
            if (updatedIds.size() > 0) {
                STDBUtil.updateBatch(tableName, targetTableName, updatedIds, connection, dataIdColumn);
            }
        }

        if (dataChangeInfos.size() > 0) {
            WeaverStatement statement = (WeaverStatement) connection.createStatement();
            try {
                String executeIdStr = StringUtils.join(executeIds.toArray(), ",");
                String updateDataChangeInfoSQL = "update "+ STDBConstants.DATA_CHANGE_TABLE_NAME +" set synced='1', sync_time = "+ System.currentTimeMillis() +" where id in (" + executeIdStr + ")";
                statement.originExecuteUpdate(updateDataChangeInfoSQL);
            } catch (Exception e) {
                throw e;
            } finally {
                if (statement != null) {
                    statement.close();
                }
            }
        }

    }

    /**
     * 获取数据变动记录
     * @param connection 数据库连接
     * @param count 获取变动数据量
     * @return 数据变动记录
     * @throws SQLException
     */
    private List<DataChangeInfo> getDataChangeInfos(WeaverConnection connection, Integer count) throws SQLException {
        List<DataChangeInfo> dataChangeInfos = new ArrayList<>();
        String sql = STContext.SYNC_SQL().selectDataChange(count);
        WeaverStatement statement = (WeaverStatement) connection.createStatement();
        try {
            ResultSet rs = statement.originExecuteQuery(sql);
            while (rs.next()){
                Long id = rs.getLong("id");
                String type = rs.getString("event_type");
                String tableName = rs.getString("table_name");
                String dataIdColumn = rs.getString("data_id_column");
                String dataId = rs.getString("data_id");
                String synced = rs.getString("synced");
                String createTime = rs.getString("create_time");
                String syncTime = rs.getString("sync_time");
                DataChangeInfo dataChangeInfo = new DataChangeInfo(id, type, tableName, dataIdColumn, dataId, synced, createTime, syncTime);
                dataChangeInfos.add(dataChangeInfo);
            }
        } catch (Exception e) {
            throw e;
        } finally {
            if (statement != null) {
                statement.close();
            }
        }
        return dataChangeInfos;
    }

    /**
     * 获取数据清理记录
     * @param connection 数据库连接
     * @param tableName 表名
     * @return 对应表的数据清理记录 只包含 id 和 dataId 两个字段
     * @throws SQLException
     */
    private Map<String, Long> getDataClearInfos(WeaverConnection connection, String tableName) throws SQLException {
        Map<String, Long> dataClearInfos = new HashMap<>();
        String sql = "select id, data_id from "+ STDBConstants.DATA_CLEAR_TABLE_NAME +" where synced = '0' and table_name='" + tableName + "'";
        WeaverStatement statement = (WeaverStatement) connection.createStatement();
        try {
            ResultSet rs = statement.originExecuteQuery(sql);
            while (rs.next()){
                Long id = rs.getLong("id");
                String dataId = rs.getString("data_id");
                dataClearInfos.put(dataId, id);
            }
        } catch (Exception e) {
            throw e;
        } finally {
            if (statement != null) {
                statement.close();
            }
        }
        return dataClearInfos;
    }

}
